package com.sh.data.engine.mqtt;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;

@Slf4j
public class EngineMqttCallback implements MqttCallbackExtended {

    private String clientId;
    private EngineMqttConsumer engineMqttConsumer;

    public EngineMqttCallback(String clientId, EngineMqttConsumer engineMqttConsumer) {
        this.clientId = clientId;
        this.engineMqttConsumer = engineMqttConsumer;
    }

    /**
     * 客户端丢失了对服务端的连接后触发的回调
     *
     * @param cause
     */
    @Override
    public void connectionLost(Throwable cause) {
        // 资源的清理  重连
        log.error("clientId {},网络或mqtt broker异常，mqtt服务器连接失败，将进行连接重试，请稍候...", clientId);
    }

    /**
     * 应用收到消息后触发的回调 该方法由mqtt客户端同步调用,在此方法未正确返回之前，不会发送ack确认消息到broker
     * 一旦该方法向外抛出了异常客户端将异常关闭，当再次连接时；所有QoS1,QoS2且客户端未进行ack确认的消息都将由 broker服务器再次发送到客户端
     *
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        try {
            String messageStr = new String(message.getPayload(), "utf-8");
            String logMessage =
                messageStr.length() > 256
                    ? StringUtils.substring(messageStr, 0, 256) + " . . ."
                    : messageStr;
            log.info(
                "clientId {},收到来自mqtt的消息,topic={},messageId={},qos={},payload=\n{}",
                clientId,
                topic,
                message.getId(),
                message.getQos(),
                logMessage);
            engineMqttConsumer.consume(messageStr);
        } catch (Exception e) {
            log.error("clientId {},handle mqtt message error", clientId, e);
        }
    }

    /**
     * 消息发布完成且收到ack确认后的回调 QoS0：消息被网络发出后触发一次 QoS1：当收到broker的PUBACK消息后触发 QoS2：当收到broer的PUBCOMP消息后触发
     *
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        int messageId = token.getMessageId();
        String[] topics = token.getTopics();
        log.info("clientId {}, 消息发布完成, messageId {}, topics {}", clientId, messageId, topics);
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("clientId {}, 连接Mqtt服务器{}成功", clientId, serverURI);
        // connect or reconnect成功后，重新进行topic订阅
        if (StrUtil.isNotBlank(engineMqttConsumer.getTopic())) {
            engineMqttConsumer.getEngineMqttClient().subscribe(engineMqttConsumer.getTopic(), 1);
        }
    }
}
